package com.infinitus.nginxInfo.config;


import java.util.Properties;

import org.springframework.beans.factory.annotation.Value;
import org.springframework.stereotype.Component;

import com.baidu.disconf.client.common.annotations.DisconfFile;
import com.baidu.disconf.client.common.annotations.DisconfFileItem;
import com.baidu.disconf.client.common.annotations.DisconfUpdateService;


/**
 * kafka配置，及相关topic
 *
 */
@Component
@DisconfFile(filename = "kafka-config.properties")
@DisconfUpdateService(classes = {KafkaConfig.class})
public class KafkaConfig {

	@Value("${clean.kafka.bootstrap.servers}")
	private String servers;
	
	@Value("${dirty.kafka.bootstrap.servers}")
	private String elkServers;
	
	@Value("${clean.kafka.acks}")
	private String acks;
	
	@Value("${clean.kafka.retries}")
	private String retries;
	
	@Value("${clean.kafka.batch.size}")
	private String size;
	
	@Value("${clean.kafka.linger.ms}")
	private String ms;
	
	@Value("${clean.kafka.buffer.memory}")
	private String memory;
	
	@Value("${clean.kafka.enable.auto.commit}")
	private String autoCommit;
	
	@Value("${clean.kafka.auto.commit.interval.ms}")
	private String commitIntervalms;
	
	@Value("${clean.session.timeout.ms}")
	private String timeout;
	
	@Value("${clean.kafka.key.serializer}")
	private String keySerializer;
	
	@Value("${clean.kafka.value.serializer}")
	private String valueSerializer;
	
	@Value("${clean.kafka.key.deserializer}")
	private String keyDeserializer;
	
	@Value("${clean.kafka.value.deserializer}")
	private String valueDeserializer;
	
	@Value("${clean.kafka.group.id}")
	private String groupId;
	
	@Value("${dirty.nginx.topic}")
	private String dirtyNginxTopic;
	
	@Value("${clean.nginx.topic}")
	private String cleanNginxTopic;
	
	
	@Value("${dirty.switchboard.topic}")
	private String dirtySwitchTopic;
	
	@Value("${clean.switchboard.topic}")
	private String cleanSwitchTopic;
	
	@Value("${userid.decryption.key}")
	private String userDecryKey;
	
	@DisconfFileItem(name = "userid.decryption.key", associateField = "userDecryKey")
	public String getUserDecryKey() {
		return userDecryKey;
	}

	
	@DisconfFileItem(name = "dirty.nginx.topic", associateField = "dirtyNginxTopic")
	public String getDirtyNginxTopic() {
		return dirtyNginxTopic;
	}

	@DisconfFileItem(name = "clean.nginx.topic", associateField = "cleanNginxTopic")
	public String getCleanNginxTopic() {
		return cleanNginxTopic;
	}

	@DisconfFileItem(name = "dirty.switchboard.topic", associateField = "dirtySwitchTopic")
	public String getDirtySwitchTopic() {
		return dirtySwitchTopic;
	}

	@DisconfFileItem(name = "clean.switchboard.topic", associateField = "cleanSwitchTopic")
	public String getCleanSwitchTopic() {
		return cleanSwitchTopic;
	}

	
	public Properties getProperties(boolean isCleanKafka, boolean isConsumer){
		Properties props = new Properties();
		if(isCleanKafka){
			props.put("bootstrap.servers", servers);
		}else{
			props.put("bootstrap.servers", elkServers);
		}
		if(!isConsumer){
			props.put("acks", acks);
			props.put("retries", retries);
			props.put("batch.size", size);
			props.put("linger.ms", ms);
			props.put("buffer.memory", memory);
			props.put("key.serializer", keySerializer);
			props.put("value.serializer", valueSerializer);
		}else{
			props.put("group.id", groupId);
			props.put("enable.auto.commit", autoCommit);
			props.put("auto.commit.interval.ms", commitIntervalms);
			props.put("session.timeout.ms", timeout);
			props.put("key.deserializer", keyDeserializer);
			props.put("value.deserializer", valueDeserializer);
		}
		return props;
	}

	
	public Properties getTestConsumerProps(String groupid){
		Properties props = new Properties();
		props.put("bootstrap.servers", elkServers);
		props.put("group.id", groupid);
		props.put("key.deserializer", keyDeserializer);
		props.put("value.deserializer", valueDeserializer);
		return props;
	}

}
